package com.edata.bigdata.flink.postgres

import com.edata.bigdata.annotations.Edata_Sink
import com.edata.bigdata.flink.Sink
import org.apache.flink.connector.jdbc.JdbcConnectionOptions.JdbcConnectionOptionsBuilder
import org.apache.flink.connector.jdbc.JdbcSink
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

import scala.reflect.ClassTag


@Edata_Sink(target = "FPSINK")
class FPSink[T: ClassTag] extends Sink[T] with FPConnector {
  override var env: StreamExecutionEnvironment = _

  override def sink(ds: DataStream[T]): Unit = {
    val clazz = implicitly[ClassTag[T]].runtimeClass

  }

  def sink(dataStream: DataStream[Array[String]], primary: String): Unit = {

    val statement = new Statement[T].setId("id").build()
    val jdbcConnOptions = new JdbcConnectionOptionsBuilder()
      .withUrl(url)
      .withDriverName(driverName)
      .withUsername(userName)
      .withPassword(password)
      .build()
    JdbcSink.sink(
      statement.sinkSQL,
      statement,
      jdbcConnOptions
    )
  }
}
